package com.gitee.melin.bee.core.jdbc.dialect.redshift;

import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.DataTypeConvertor;
import com.gitee.melin.bee.core.jdbc.dialect.DefaultTypeConvertor;
import com.gitee.melin.bee.core.jdbc.dialect.IDBQuery;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.List;

public class RedshiftJdbcDialect extends AbstractJdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(RedshiftJdbcDialect.class);

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.POSTGRESQL;
    }

    private static final String QUERY_DATABASES_SQL =
            "select datname from pg_catalog.pg_database where datistemplate = false";

    private static final String VERSION_SQL = "SELECT version()";

    private static final double SUPPORT_RELPARTBOUND_START_VERSION = 10;

    private static final String SUPPORT_RELPARTBOUND_STATEMENT = "AND c.relpartbound IS NULL";

    private static final String NOT_SUPPORT_RELPARTBOUND_STATEMENT =  "and not EXISTS(SELECT 1 from pg_inherits  where pg_inherits.inhrelid = c.oid)";

    private Integer isHighVersion = 0;//高版本支持relpartbound字段:1,低于10下版本不支持 -1

    public RedshiftJdbcDialect(ConnectionInfo connectionInfo) {
        super(connectionInfo);
    }

    public RedshiftJdbcDialect(Connection connection) {
        super(connection);
    }

    @Override
    protected IDBQuery getDBQuery() {
        //判断数据库版本,高版本支持relpartbound字段,低版本不支持
        this.sqlCompatibleVersion();
        if (isHighVersion > 0) {
            return new RedshiftQuery(SUPPORT_RELPARTBOUND_STATEMENT);
        }
        return new RedshiftQuery(NOT_SUPPORT_RELPARTBOUND_STATEMENT);
    }

    private void sqlCompatibleVersion() {
        if (isHighVersion != 0) {
            return;
        }

        try (Connection connection = this.getConnection();
             Statement statement = connection.createStatement();
             ResultSet versionRs = statement.executeQuery(VERSION_SQL)){
            String version = null;
            while (versionRs.next()) {
                version = versionRs.getString(1);
            }

            if (version.contains("(")) {
                version = version.substring(0, version.indexOf("("));
                version = version.replaceFirst("PostgreSQL", "").trim();
            }

            if (version.contains(".")){
                version = version.split("\\.")[0];
            }
            double versionNumber = Double.parseDouble(version);
            if (versionNumber > SUPPORT_RELPARTBOUND_START_VERSION) {
                isHighVersion = 1;
            } else {
                isHighVersion = -1;
            }
        } catch (Exception ex) {
            isHighVersion = 1;
            LOG.error(ex.getMessage(), ex);
        }
    }

    @Override
    protected DataTypeConvertor getTypeConvert() {
        return new DefaultTypeConvertor();
    }

    @Override
    protected Driver getDriver() throws SQLException {
        return loadDriver(DataSourceType.REDSHIFT.getDriverClass());
    }

    @Override
    protected boolean filterSchema(String schemaName) {
        if (schemaName.equalsIgnoreCase("pg_catalog")) {
            return false;
        }
        return super.filterSchema(schemaName);
    }

    @Override
    protected String getDDLColumnName() {
        return "Show Table DDL statement";
    }

    @Override
    public List<String> getDatabases() {
        try (Connection connection = this.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(QUERY_DATABASES_SQL);
             ResultSet resultSet = preparedStatement.executeQuery()) {

            List<String> catalogs = Lists.newArrayList();
            while (resultSet.next()) {
                String catalogName = resultSet.getString(1);
                catalogs.add(catalogName);
            }
            return catalogs;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    protected String createTableSql(String schemaName, String tableName) {
        return "SHOW TABLE \""+ schemaName +"\".\""+ tableName +"\"";
    }
}
